package org.zezs.sync.mongo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoIterable;
import jdk.nashorn.api.scripting.NashornScriptEngine;
import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.bson.types.ObjectId;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zezs.sync.mongo.config.es.EsUtil;
import org.zezs.sync.mongo.config.mongo.MongoUtil;
import org.zezs.sync.mongo.config.sync.SyncUtil;
import org.zezs.sync.mongo.config.sync.metadata.*;
import org.zezs.sync.mongo.plugs.Plug;
import org.zezs.sync.mongo.utils.BlockThreadPool;
import org.zezs.sync.mongo.utils.InputStreamUtil;

import javax.script.ScriptEngineManager;
import javax.script.ScriptException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.charset.Charset;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RunTimeUtil {

    private static Logger logger = LoggerFactory.getLogger(RunTimeUtil.class);

    private ScriptEngineManager manager = new ScriptEngineManager();

    private NashornScriptEngine engine = null;

    private static Map<String, RunTimeUtil> runTimeUtil = new HashMap<>();


    static BlockThreadPool threadPoolExecutor = new BlockThreadPool(1);

    private RunTimeUtil() {

    }

    public static RunTimeUtil getJsRunTimeUtil(SyncEnt syncEnt) {
        String key = String.valueOf(syncEnt.hashCode());
        RunTimeUtil jsRunTimeUtil = runTimeUtil.get(key);
        if (jsRunTimeUtil == null) {
            jsRunTimeUtil = new RunTimeUtil();
            jsRunTimeUtil.engine = (NashornScriptEngine) jsRunTimeUtil.manager.getEngineByName("javascript");
            /*加载js脚本*/
            try {
                byte[] bytes = InputStreamUtil.readAll("/loadjs.json");
                List<String> parse = JSONArray.parseArray(new String(bytes, Charset.forName("utf-8")), String.class);
                for (int i = 0; i < parse.size(); i++) {
                    Reader core = new InputStreamReader(RunTimeUtil.class.getResourceAsStream(parse.get(i)));
                    try {
                        jsRunTimeUtil.engine.eval(core);
                    } catch (ScriptException e) {
                        e.printStackTrace();
                    }
                }
                runTimeUtil.put(key, jsRunTimeUtil);
            } catch (Exception e) {
                logger.error("加载js预制脚本失败", e);
            }

            try {
                String[] plugs = syncEnt.getPlugs();
                if (plugs != null && plugs.length > 0) {
                    for (String plug : plugs) {
                        if (StringUtils.isNotBlank(plug)) {
                            Class<?> aClass = Class.forName(plug);
                            Plug plugObj = (Plug) aClass.newInstance();
                            plugObj.init(jsRunTimeUtil.engine);
                        }
                    }

                }
            } catch (Exception e) {
                logger.error("加载js插件失败", e);
            }

        }

        return jsRunTimeUtil;

    }


    /**
     * 初始化同步
     */
    public static void initSync() {

        List<SyncEnt> syncEnts = SyncUtil.init();
        for (SyncEnt syncEnt : syncEnts) {
            if (syncEnt.isUse()) {
                Thread thread = new Thread(() -> {
                    try {
                        getJsRunTimeUtil(syncEnt);
                        syncObj(syncEnt);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });

                if (StringUtils.isNotBlank(syncEnt.getName())) {
                    thread.setName(syncEnt.getName());
                }
                thread.start();

            }

        }
    }




    /**
     * 初始化同步对象
     *
     * @param syncEnt 同步对象
     * @throws ParseException
     * @throws ScriptException
     * @throws InterruptedException
     */
    public static void syncObj(SyncEnt syncEnt) throws ParseException, ScriptException, InterruptedException {
        logger.info(JSON.toJSONString(syncEnt));
        MongoIterable<String> listCollectionNames = MongoUtil.getMongoDb(syncEnt.getSource()).listCollectionNames();
        List<String> cols = new ArrayList<>();
        MongoCursor<String> iterator = listCollectionNames.iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            if (next.matches(syncEnt.getCollection())) {
                cols.add(next);
            }
        }
        Collections.sort(cols);

        List<Rules> totalRules = new ArrayList<>();

        List<Chains> chains = syncEnt.getChains();
        for (Chains chain : chains) {
            chain.setRef_syncEnt(syncEnt);
            getJsRunTimeUtil(syncEnt).engine
                    .eval("apiScript['" + chain.hashCode() + "']=function(obj){" + chain.getJsscript() + "}");
            logger.info("初始化函数  run {}", JSON.toJSONString(chain));

            List<Rules> rules = chain.getRules();
            for (Rules rule : rules) {
                rule.setRef_chains(chain);

                if (!rule.isUse()) {
                    continue;
                }

                totalRules.add(rule);

                String maxMongoId = getDestLastMongoId(rule, chain);

                initSourceRule(syncEnt, cols, rule, maxMongoId);
            }
        }

        //根据最小值对所有rule数据排序
        Collections.sort(totalRules, (Rules o1, Rules o2) -> {
            if (o1.getMin() == o2.getMin()) {
                return 0;
            } else if (o2.getMin() == null) {
                return 1;
            } else if (o1.getMin() == null) {
                return -1;
            }
            return o1.getMin().toString().compareTo(o2.getMin().toString());
        });
        String min_max_mongoId = null;
        String min_max_Collection = null;
        for (Rules rules : totalRules) {
            if (min_max_mongoId == null && rules.getMin() != null) {
                rules.setWait(false);
                min_max_mongoId = rules.getMin();
                min_max_Collection = rules.getMinCollection();
            } else {
                rules.setWait(true);
            }
        }
        if (cols.isEmpty()) {
            logger.info("没找到任何匹配规则集合 {}", syncEnt.getCollection());
            return;
        }

        if (min_max_mongoId != null) {
            String finalMin_max_mongoId = min_max_mongoId;
            String finalMin_max_Collection = min_max_Collection;
            cols.forEach(col -> {
                totalRules.forEach(rule -> {
                    if (rule.getMinCollection().compareTo(col) > 0) {
                        logger.info("当前集合已经同步，跳过 {} ", col);
                    } else if (col.compareTo(finalMin_max_Collection) <= 0) {
                        syncColl(rule, finalMin_max_mongoId, finalMin_max_Collection, col);
                    } else {

                    }
                });
            });
            syncMaster(syncEnt, new ObjectId(min_max_mongoId), min_max_Collection, true);
        } else {

            syncMaster(syncEnt, null, cols.get(0), true);
        }
    }

    private static String getDestLastMongoId(Rules rule, Chains chain) {
        if ("mongo".equals(chain.getDestType())) {

            BasicDBObject destWhere = new BasicDBObject();
            List<Where> destWheres = rule.getDestWhere();
            for (Where where1 : destWheres) {
                destWhere.append(where1.getName(), where1.getValue());
            }

            MongoCollection<Document> collection = MongoUtil.getMongoDb(chain.getDest())
                    .getCollection(chain.getTable());
            BasicDBObject sort = new BasicDBObject(Constants.mongoId, -1);//目的表最后一条数据
            FindIterable<Document> limit = collection.find(destWhere).sort(sort).limit(1);

            Document first = limit.first();
            if (first != null && first.get(Constants.mongoId) != null) {
                return first.get(Constants.mongoId).toString();
            }
        }
        if ("es".equals(chain.getDestType())) {

            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            List<QueryBuilder> must = boolQuery.must();
            List<Where> destWheres = rule.getDestWhere();
            for (Where destWhere : destWheres) {
                must.add(QueryBuilders.termQuery(destWhere.getName(), destWhere.getValue()));
            }
            TransportClient client = EsUtil.getClient(chain.getDest());
            SearchRequestBuilder searchRequestBuilder = client.prepareSearch(chain.getDb())
                    .setTypes(chain.getTable()).setFetchSource(new String[]{Constants.mongoId}, null).setSize(1).addSort(Constants.mongoId + ".keyword", SortOrder.DESC);
            searchRequestBuilder.setQuery(boolQuery);
            SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
            if (searchResponse.getHits().getTotalHits() > 0) {
                SearchHit at = searchResponse.getHits().getAt(0);
                Map<String, Object> sourceAsMap = at.getSourceAsMap();
                Object mongoId = sourceAsMap.get(Constants.mongoId);
                if (mongoId != null) {

                    return mongoId.toString();
                }
            }
        }
        return null;
    }


    private static void initSourceRule(SyncEnt syncEnt, List<String> cols, Rules rule, String mongoId) {
        List<Where> where = rule.getWhere();
        JSONObject sourceQuery = new JSONObject();
        for (Where where2 : where) {
            sourceQuery.put(where2.getName(), where2.getValue());
        }
        for (Where where2 : syncEnt.getWhere()) {
            sourceQuery.put(where2.getName(), where2.getValue());
        }
        Document lastSource;
        if (mongoId != null) {
            BasicDBObject basicDBObject = new BasicDBObject("_id", new BasicDBObject("$lte", new ObjectId(mongoId)));
            basicDBObject.putAll(sourceQuery);
            for (String collectionName : cols) {
                MongoCollection<Document> collection2 = MongoUtil.getMongoDb(syncEnt.getSource())
                        .getCollection(collectionName);
                FindIterable<Document> limit2 = collection2.find(basicDBObject)
                        .sort(new BasicDBObject("_id", -1)).limit(1);
                lastSource = limit2.first();
                if (lastSource != null) {
                    rule.setMin(lastSource.get("_id").toString());
                    rule.setMinRow(lastSource);
                    rule.setMinCollection(collectionName);
                    rule.setWait(true);
                }
            }
        }
    }


    private static void syncColl(Rules rule, String max, String maxCollectionName, String collectionName) {
        SyncEnt syncEnt = rule.getRef_chains().getRef_syncEnt();
        MongoCollection<Document> collection2 = MongoUtil.getMongoDb(syncEnt.getSource()).getCollection(collectionName);

        BasicDBObject query = new BasicDBObject();

        for (Where where2 : rule.getWhere()) {
            query.append(where2.getName(), where2.getValue());
        }
        for (Where where2 : syncEnt.getWhere()) {
            query.put(where2.getName(), where2.getValue());
        }

        if (rule.getMin() != null) {
            if (rule.getMinCollection().compareTo(collectionName) > 0) {
                return;
            }

            BasicDBObject idRang = new BasicDBObject("$gt", new ObjectId(rule.getMin()));
            if (collectionName.equals(maxCollectionName)) {
                idRang.put("$lte", new ObjectId(max));
            }
            query.put("_id", idRang);
        }

        FindIterable<Document> limit1 = collection2.find(query).limit(syncEnt.getLimit());
        Document first = limit1.first();
        if (first == null) {
            logger.info("已经完成历史数据同步 {}", collectionName);
        } else {
            long total = collection2.count(query);

            String begin = rule.getMin();
            while (begin != null) {
                query = new BasicDBObject();

                for (Where where2 : rule.getWhere()) {
                    query.append(where2.getName(), where2.getValue());
                }
                for (Where where2 : syncEnt.getWhere()) {
                    query.put(where2.getName(), where2.getValue());
                }

                if (rule.getMin() != null) {
                    if (rule.getMinCollection().compareTo(collectionName) > 0) {
                        return;
                    }

                    BasicDBObject idRang = new BasicDBObject("$gt", new ObjectId(rule.getMin()));
                    if (collectionName.equals(maxCollectionName)) {
                        idRang.put("$lte", new ObjectId(max));
                    }
                    query.put("_id", idRang);
                }
                long querBegin = System.currentTimeMillis();
                FindIterable<Document> findIterable = collection2.find(query).sort(new BasicDBObject("_id", 1)).limit(syncEnt.getLimit());

                if (findIterable.first() == null) {
                    begin = null;
                }
                int count = 0;
                MongoCursor<Document> iterator2 = findIterable.iterator();
                long querEnd = System.currentTimeMillis();
                List<Document> queryData = new ArrayList<>(syncEnt.getLimit());
                while (iterator2.hasNext()) {
                    Document next = iterator2.next();
                    begin = next.get("_id").toString();
                    queryData.add(next);

                }
                List<JSONObject> jsonObjects = rule.getRef_chains().getProesser().proess(rule.getRef_chains(), queryData);

                total -= count;
                long jsExtime = System.currentTimeMillis();

                String last = "";
                if (begin != null) {
                    DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    last = dateFormat.format(new ObjectId(begin).getDate());
                }
                ProcessInfo processInfo = new ProcessInfo(querEnd - querBegin, jsExtime - querEnd, count, last);
                processInfo.setPreNum(total);
                rule.getRef_chains().setProcessInfo(processInfo);

                doBatchInsert(rule.getRef_chains(), jsonObjects, collectionName);

            }

        }

    }

    private static void doBatchInsert(Chains chains, List<JSONObject> batchData, String collectionName) {
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                SyncEnt ref_syncEnt = chains.getRef_syncEnt();
                long begin = System.currentTimeMillis();
                if (!batchData.isEmpty()) {
                    if (chains.getDestType().equals("mongo")) {
                        MongoCollection<Document> collection = MongoUtil.getMongoDb(chains.getDest())
                                .getCollection(chains.getTable());
                        List<Document> data = new ArrayList<>();
                        batchData.forEach(action -> data.add(new Document(action)));
                        collection.insertMany(data);
                    } else {
                        TransportClient client = EsUtil.getClient(chains.getDest());
                        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
                        for (int i = 0; i < batchData.size(); i++) {
                            JSONObject obj = batchData.get(i);
                            IndexRequest request = client
                                    .prepareIndex(chains.getDb(), chains.getTable()).setSource(obj)
                                    .request();
                            bulkRequestBuilder.add(request);
                        }
                        ActionFuture<BulkResponse> execute = bulkRequestBuilder.execute();
                        BulkResponse bulkResponse = execute.actionGet();
                        if (bulkResponse.hasFailures()) {

                            logger.error("{},同步es错误错误！", chains.getRef_syncEnt().getSource() + "." + collectionName + "-->" + chains.getDest() + "." + chains.getTable());
                            logger.error(bulkResponse.buildFailureMessage());
                        }
                    }
                }

                long cost = System.currentTimeMillis() - begin;

                if (chains.getProcessInfo().getPreNum() == null) {
                    logger.info("{} 查询/处理/插入 {}/{}/{}，行数 {}， 最新记录{}",
                            ref_syncEnt.getSource() + "." + collectionName + "-->" + chains.getDest() + "." + chains.getTable(),
                            chains.getProcessInfo().getQueryTime(), chains.getProcessInfo().getJstime(), cost, batchData.size(), chains.getProcessInfo().getLastTime());
                } else {
                    logger.info("{} 查询/处理/插入 {}/{}/{}，行数{}，剩余{} 最新记录{}",
                            ref_syncEnt.getSource() + "." + collectionName + "-->" + chains.getDest() + "." + chains.getTable(),
                            chains.getProcessInfo().getQueryTime(), chains.getProcessInfo().getJstime(), cost, batchData.size(),
                            chains.getProcessInfo().getPreNum(), chains.getProcessInfo().getLastTime());
                }
            }
        });


    }

    public static void syncMaster(SyncEnt syncEnt, ObjectId min, String currColl, boolean doCount) {
        boolean flag = true;
        Long total = null;
        while (flag) {
            MongoCollection<Document> collection2 = MongoUtil.getMongoDb(syncEnt.getSource()).getCollection(currColl);
            Date date = new Date();
            int l = (int) (date.getTime() / 1000 - 300);
            ObjectId max = new ObjectId(l, 0, (short) 0, 0);
            BasicDBObject query = new BasicDBObject();

            for (Where where2 : syncEnt.getWhere()) {
                query.put(where2.getName(), where2.getValue());
            }
            BasicDBObject idWh = new BasicDBObject("$lte", max);
            if (min != null) {
                idWh.put("$gt", min);
            }
            query.put("_id", idWh);

            if (doCount) {
                total = collection2.count(query);
                doCount = false;
            }

            long querBegin = System.currentTimeMillis();
            FindIterable<Document> findIterable = collection2.find(query).sort(new BasicDBObject("_id", 1)).limit(syncEnt.getLimit());

            MongoCursor<Document> iterator2 = findIterable.iterator();
            long querEnd = System.currentTimeMillis();
            int count = 0;
            List<Document> querResult = new ArrayList<>(syncEnt.getLimit());
            while (iterator2.hasNext()) {
                Document next = iterator2.next();
                min = next.getObjectId("_id");
                count++;
                querResult.add(next);
            }

            String last = "";
            if (min != null) {
                DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                last = dateFormat.format(min.getDate());
            }
            total = total - count;
            for (Chains chains : syncEnt.getChains()) {
                querEnd = System.currentTimeMillis();
                List<JSONObject> batchData = chains.getProesser().proess(chains, querResult);
                long jsExtime = System.currentTimeMillis();
                ProcessInfo processInfo = new ProcessInfo(querEnd - querBegin, jsExtime - querEnd, count, last);
                if (total > 0) {
                    processInfo.setPreNum(total);
                }
                chains.setProcessInfo(processInfo);
                doBatchInsert(chains, batchData, currColl);
            }

            if (count >= syncEnt.getLimit()) {
                syncEntCount.put(syncEnt.hashCode(), new AtomicInteger());
            } else {
                AtomicInteger atomicInteger = syncEntCount.get(syncEnt.hashCode());
                if (atomicInteger == null) {
                    atomicInteger = new AtomicInteger();
                    syncEntCount.put(syncEnt.hashCode(), atomicInteger);
                }
                int i = atomicInteger.incrementAndGet();
                if (i > 5) {
                    atomicInteger.set(5);
                    MongoIterable<String> listCollectionNames = MongoUtil.getMongoDb(syncEnt.getSource()).listCollectionNames();
                    List<String> cols = new ArrayList<>();
                    MongoCursor<String> iterator = listCollectionNames.iterator();
                    while (iterator.hasNext()) {
                        String next = iterator.next();
                        if (next.matches(syncEnt.getCollection())) {
                            cols.add(next);
                        }
                    }
                    Collections.sort(cols);
                    int index = cols.indexOf(currColl);
                    if (index < cols.size() - 1) {
                        currColl = cols.get(index + 1);
                        min = null;
                        doCount = true;
                    }
                }
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }


        }
    }


    static Map<Integer, AtomicInteger> syncEntCount = new HashMap<>();


    public static Object getDest(Mappings lastField, Object object) {
        Date parse;
        try {
            if (object == null) {
                return lastField.getValue();
            } else if (lastField.getSourceType() != lastField.getDestType()) {
                if ("string".equals(lastField.getDestType()) && "date".equals(lastField.getSourceType())) {
                    DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    String format = dateFormat.format(object);
                    return format;
                }
            } else if ("date".equals(lastField.getDestType()) && "string".equals(lastField.getSourceType())) {
                DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                parse = dateFormat.parse(object.toString());

                return parse;
            }
        } catch (ParseException e) {
            return null;
        }
        return object;
    }

    public NashornScriptEngine getEngine() {
        return engine;
    }

    public void setEngine(NashornScriptEngine engine) {
        this.engine = engine;
    }
}
